package minicloud;

import akka.Done;
import akka.actor.typed.ActorSystem;
import akka.actor.typed.javadsl.Behaviors;
import akka.stream.javadsl.*;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.CompletionStage;

// 日志数据模型
class LogEntry {
    final long id;
    final String level;
    final String message;
    final LocalDateTime createTime;
    final String taskId;

    public LogEntry(long id, String level, String message, LocalDateTime createTime,String taskId) {
        this.id = id;
        this.level = level;
        this.message = message;
        this.createTime = createTime;
        this.taskId = taskId;
    }
}

// 主处理类
public class MysqlStreamProcessor {
    private final MysqlPersistence persistence;
    private final String taskId;

    // 构造函数传入taskId，指定当前任务
    public MysqlStreamProcessor(String taskId) {
        this.taskId = taskId;
        this.persistence = new MysqlPersistence(taskId);
    }

    public void start() {
        ActorSystem<Void> system = ActorSystem.create(Behaviors.empty(), "StreamSystem-" + taskId);

        // 数据源：只读取当前taskId的日志
        Source<LogEntry, ?> mysqlSource = Source.tick(
                        Duration.ZERO,
                        Duration.ofSeconds(5),
                        "pull-logs"
                )
                .flatMapConcat(tick -> {
                    long lastId = Long.parseLong(persistence.getState("last_processed_id"));
                    List<LogEntry> unprocessed = persistence.readUnprocessedLogs(lastId, 10);
                    System.out.println("任务[" + taskId + "]拉取到 " + unprocessed.size() + " 条日志");
                    return Source.from(unprocessed);
                });

        // 处理流：逻辑同前，但所有操作绑定到当前taskId
        // 2. 处理流：过滤ERROR日志 + 窗口聚合 + 状态更新（完整逻辑）
        /**
         * 完整逻辑说明
         * 窗口判断与切换
         * 每次处理日志时，先计算当前窗口开始时间（如 10:00:00），并与数据库中存储的 current_window_start 对比。
         * 若时间不一致（进入新窗口），则先保存上一个窗口的统计结果，再初始化新窗口的计数（从 1 开始，当前日志属于新窗口）。
         * 状态持久化细节
         * 当前窗口计数：每处理一条日志就更新 current_window_count 到数据库，确保宕机后重启能恢复累计值。
         * 已处理日志 ID：仅当当前日志 ID 大于 last_processed_id 时才更新，避免重复处理（幂等性核心）。
         * 窗口结束兜底机制
         * 即使因异常导致窗口未自动触发（如 groupedWithin 未执行），在每次新窗口切换时也会主动保存上一个窗口的结果，双重保障数据不丢失。
         * 多任务适配
         * 所有状态操作（读 / 写 processing_state 表）均通过绑定 taskId 的 persistence 实例完成，确保不同任务的状态隔离。
         * 此逻辑覆盖了单条日志处理、窗口切换、状态持久化和幂等性保障，可直接集成到多任务并行场景中，配合之前的表结构和测试用例验证正确性。
         */
        Flow<LogEntry, String, ?> processingFlow = Flow.of(LogEntry.class)
                // 步骤1：过滤出ERROR级别日志（仅处理当前任务的ERROR日志）
                .filter(log -> "ERROR".equals(log.level))
                // 步骤2：逐条处理日志，更新状态（核心逻辑）
                .map(log -> {
                    // a. 获取当前窗口开始时间（如 10:00:00，整分钟对齐）
                    LocalDateTime currentWindowStart = LocalDateTime.now()
                            .withSecond(0).withNano(0); // 对齐到分钟级窗口

                    // b. 从数据库读取当前任务的状态（当前窗口累计数、上次窗口开始时间）
                    String storedWindowStartStr = persistence.getState("current_window_start");
                    LocalDateTime storedWindowStart = storedWindowStartStr.isEmpty()
                            ? null
                            : LocalDateTime.parse(storedWindowStartStr, MysqlPersistence.dtf);
                    int currentWindowCount = Integer.parseInt(persistence.getState("current_window_count"));

                    // c. 判断是否进入新窗口（当前时间已超过存储的窗口开始时间）
                    boolean isNewWindow = storedWindowStart == null
                            || !currentWindowStart.isEqual(storedWindowStart);

                    if (isNewWindow) {
                        // 若进入新窗口：
                        // 1. 先保存上一个窗口的统计结果（如果存在）
                        if (storedWindowStart != null) {
                            persistence.saveStatistics(storedWindowStart, currentWindowCount);
                            System.out.println("任务[" + taskId + "] 窗口 " + storedWindowStart + " 已保存，计数：" + currentWindowCount);
                        }
                        // 2. 初始化新窗口状态
                        currentWindowCount = 1; // 当前日志属于新窗口，计数从1开始
                        persistence.updateState("current_window_start", currentWindowStart.format(MysqlPersistence.dtf));
                    } else {
                        // 若仍在当前窗口：累计计数
                        currentWindowCount++;
                    }

                    // d. 更新当前窗口累计数（持久化到数据库，防止宕机丢失）
                    persistence.updateState("current_window_count", String.valueOf(currentWindowCount));

                    // e. 更新已处理的最大日志ID（确保幂等性：仅当当前日志ID更大时才更新）
                    long lastProcessedId = Long.parseLong(persistence.getState("last_processed_id"));
                    if (log.id > lastProcessedId) {
                        persistence.updateState("last_processed_id", String.valueOf(log.id));
                    }

                    // 返回中间结果（用于后续窗口聚合输出）
                    return new WindowResult(currentWindowStart, currentWindowCount);
                })
                // 步骤3：窗口结束触发（每1分钟），输出最终统计结果
                .groupedWithin(Integer.MAX_VALUE, Duration.ofMinutes(1))
                .map(windowResults -> {
                    if (windowResults.isEmpty()) {
                        return "任务[" + taskId + "] 窗口无ERROR日志";
                    }
                    // 取窗口内最后一条结果（包含最新累计数）
                    WindowResult lastResult = windowResults.get(windowResults.size() - 1);
                    LocalDateTime windowStart = lastResult.windowStart;
                    int totalCount = lastResult.count;

                    // 再次保存窗口结果（双重保障：即使窗口未触发自动提交，此处也会兜底）
                    persistence.saveStatistics(windowStart, totalCount);

                    // 重置当前窗口状态（准备下一个窗口）
                    persistence.updateState("current_window_count", "0");
                    persistence.updateState("current_window_start", "");

                    return String.format(
                            "任务[%s] 窗口 %s 统计完成，ERROR日志总数：%d条",
                            taskId, windowStart.format(MysqlPersistence.dtf), totalCount
                    );
                });

        // 输出端：打印任务标识
        Sink<String, CompletionStage<Done>> consoleSink = Sink.foreach(
                result -> System.out.println("任务[" + taskId + "]：" + result)
        );

        // 启动流
        mysqlSource.via(processingFlow).runWith(consoleSink, system);
    }

    // 窗口统计结果模型
    static class WindowResult {
        LocalDateTime windowStart;
        int count;

        WindowResult(LocalDateTime windowStart, int count) {
            this.windowStart = windowStart;
            this.count = count;
        }
    }

    // 启动多个并行任务
    public static void main(String[] args) {
        // 任务1：处理订单日志
        new Thread(() -> new MysqlStreamProcessor("order-log").start()).start();
        // 任务2：处理用户日志
        new Thread(() -> new MysqlStreamProcessor("user-log").start()).start();
        // 可根据需要启动更多任务...
    }
}
